package com.nageoffer.shortlink.project.mq.consumer;


import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.nageoffer.shortlink.common.constant.RedisConstant;
import com.nageoffer.shortlink.common.constant.ShortLinkConstant;
import com.nageoffer.shortlink.common.dto.project.biz.ShortLinkStatsRecordDTO;
import com.nageoffer.shortlink.project.dao.entity.*;
import com.nageoffer.shortlink.project.dao.mapper.*;
import com.nageoffer.shortlink.project.mq.idempotent.MQIdempotentHandler;
import com.nageoffer.shortlink.project.mq.producer.DelayShortLinkStatsProducer;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.*;

@Component
@Slf4j
@RequiredArgsConstructor
public class RedisMQShortLinkStatsConsumer implements StreamListener<String, MapRecord<String,String,String>> {
    private final LinkAccessStatsMapper linkAccessStatsMapper;
    private final LinkLocaleStatsMapper linkLocaleStatsMapper;
    private final LinkOsStatsMapper linkOsStatsMapper;
    private final LinkBrowserStatsMapper linkBrowserStatsMapper;
    private final LinkAccessLogsStatsMapper linkAccessLogsStatsMapper;
    private final LinkDeviceStatsMapper linkDeviceStatsMapper;
    private final LinkNetworkStatsMapper linkNetworkStatsMapper;
    private final DelayShortLinkStatsProducer delayShortLinkStatsProducer;
    private final RedissonClient redissonClient;
    private final ShortLinkGotoMapper shortLinkGotoMapper;
    private final StringRedisTemplate stringRedisTemplate;
    private final MQIdempotentHandler mqIdempotentHandler;
    @Value("${short-link.stats.locale.amap-key}")
    private String localeAmapKey;
    @Override
    public void onMessage(MapRecord<String,String,String> message){
        String stream = message.getStream();
        RecordId id = message.getId();
        if (mqIdempotentHandler.isMessageProcessed(id.toString())) {
            return;
        }
        try{
            Map<String, String> producerMap = message.getValue();
            String fullShortUrl = producerMap.get("fullShortUrl");
            if (StrUtil.isNotBlank(fullShortUrl)) {
                String gid = producerMap.get("gid");
                ShortLinkStatsRecordDTO statsRecord = JSONUtil.toBean(producerMap.get("statsRecord"), ShortLinkStatsRecordDTO.class);
                actualSaveShortLinkStats(fullShortUrl,gid,statsRecord);
            }
        }catch(Throwable ex){
            mqIdempotentHandler.delMessageProcessed(id.toString());
            log.error("记录短链接监控消费异常");
        }
        stringRedisTemplate.opsForStream().delete(Objects.requireNonNull(stream),id.getValue());
    }
    public void actualSaveShortLinkStats(String fullShortUrl,String gid,ShortLinkStatsRecordDTO statsRecord){
        /**
         * uv统计
         */
        fullShortUrl = Optional.ofNullable(fullShortUrl).orElse(statsRecord.getFullShortUrl());
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock((RedisConstant.LOCK_GID_UPDATE_KEY) + fullShortUrl);
        RLock readLock = readWriteLock.readLock();
        if (!readLock.tryLock()) {
            delayShortLinkStatsProducer.send(statsRecord);
            return;
        }
        try{
            if (gid==null) {
                LambdaQueryWrapper<ShortLinkGoto> linkGotoWrapper = Wrappers.lambdaQuery(ShortLinkGoto.class)
                        .eq(ShortLinkGoto::getFullShortUrl, fullShortUrl);
                ShortLinkGoto shortLinkGoto = shortLinkGotoMapper.selectOne(linkGotoWrapper);
                gid = shortLinkGoto.getGid();
            }
            Date date = new Date();
            Week week = DateUtil.dayOfWeekEnum(date);
            int weekValue = week.getIso8601Value();
            int hour = DateUtil.hour(date, true);
            LinkAccessStats stats = LinkAccessStats.builder()
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .uv(statsRecord.getUvFirstFlag()?1:0)
                    .pv(1)
                    .uip(statsRecord.getUipFirstFlag()?1:0)
                    .hour(hour)
                    .date(date)
                    .weekday(weekValue)
                    .build();
            linkAccessStatsMapper.shortLinkStats(stats);
            /**
             * locale统计
             */
            HashMap<String, Object> localeParamMap = new HashMap<>();
            localeParamMap.put("key",localeAmapKey);
            localeParamMap.put("ip", statsRecord.getRemoteAddr());
            String localeResult = HttpUtil.get(ShortLinkConstant.AMAP_REMOTE_URL, localeParamMap);
            JSONObject localeJson = JSONUtil.parseObj(localeResult);
            String infoCode = localeJson.getStr("infocode");
            String province = null;
            String city=null;
            boolean unKnown=true;
            if (StrUtil.isNotBlank(infoCode) && StrUtil.equals(infoCode,"10000")) {
                province = localeJson.getStr("province");
                city =localeJson.getStr("city");
                unKnown = StrUtil.isBlank(province) || "[]".equals(province);
                LinkLocaleStats localeStats = LinkLocaleStats.builder()
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .country("中国")
                        .province(unKnown?"未知":province)
                        .city(unKnown?"未知":city)
                        .adcode(unKnown?"未知":localeJson.getStr("adcode"))
                        .date(new Date())
                        .cnt(1)
                        .build();
                linkLocaleStatsMapper.shortLinkLocaleStats(localeStats);
            }
            /**
             * os统计
             */
            LinkOsStats osStats = LinkOsStats.builder()
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .date(new Date())
                    .cnt(1)
                    .os(statsRecord.getOs())
                    .build();
            linkOsStatsMapper.shortLinkOsStats(osStats);
            /**
             * browser统计
             */
            LinkBrowserStats browserStats = LinkBrowserStats.builder()
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .date(new Date())
                    .cnt(1)
                    .browser(statsRecord.getBrowser())
                    .build();
            linkBrowserStatsMapper.shortLinkBrowserStats(browserStats);
            /**
             * 访问设备统计
             */
            LinkDeviceStats deviceStats = LinkDeviceStats.builder()
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .date(new Date())
                    .cnt(1)
                    .device(statsRecord.getDevice())
                    .build();
            linkDeviceStatsMapper.shortLinkDeviceStats(deviceStats);
            /**
             * 访问网络统计
             */
            LinkNetworkStats networkStats = LinkNetworkStats.builder()
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .date(new Date())
                    .cnt(1)
                    .network(statsRecord.getNetwork())
                    .build();
            linkNetworkStatsMapper.shortLinkNetworkStats(networkStats);
            /**
             * 访问日志统计
             */
            LinkAccessLogsStats accessLogsStats = LinkAccessLogsStats.builder()
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .ip(statsRecord.getRemoteAddr())
                    .user(statsRecord.getUv())
                    .os(statsRecord.getOs())
                    .browser(statsRecord.getBrowser())
                    .network(statsRecord.getNetwork())
                    .device(statsRecord.getDevice())
                    .locale(StrUtil.join("-","中国",unKnown?"未知":province,unKnown?"未知":city))
                    .build();
            linkAccessLogsStatsMapper.insert(accessLogsStats);
        }finally {
            readLock.unlock();
        }
    }
}
